package com.atguigu.flink.chapter07;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/12/15 9:26
 */
public class Flink07_Window_ProcessFunction_2 {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env
                .socketTextStream("hadoop162", 9999)
                .map(line ->{
                    String[] strings = line.split(",");
                    return new WaterSensor(
                            strings[0],
                            Long.valueOf(strings[1]),
                            Integer.valueOf(strings[2]));
                })

                .keyBy(WaterSensor::getId)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
               .aggregate(new AggregateFunction<WaterSensor, MyAvg, Double>() {
                              @Override
                              public MyAvg createAccumulator() {
                                  System.out.println("Flink07_Window_ProcessFunction_2.createAccumulator");
                                  return new MyAvg();
                              }

                              @Override
                              public MyAvg add(WaterSensor value, MyAvg accumulator) {
                                  System.out.println("Flink07_Window_ProcessFunction_2.add");
                                  accumulator.sum += value.getVc();
                                  accumulator.count++;
                                  return accumulator;
                              }

                              @Override
                              public Double getResult(MyAvg accumulator) {
                                  System.out.println("Flink07_Window_ProcessFunction_2.getResult");
                                  return accumulator.avg();
                              }

                              @Override
                              public MyAvg merge(MyAvg a, MyAvg b) {
                                  System.out.println("Flink07_Window_ProcessFunction_2.merge");
                                  return null;
                              }
                          }, new ProcessWindowFunction<Double, String, String, TimeWindow>() {
                              @Override
                              public void process(String key, Context context,
                                                  Iterable<Double> elements,
                                                  Collector<String> out) throws Exception {
                                  System.out.println(key+"  "+context.window()+elements.iterator().next());
                              }
                          }
               )
                .print();


        env.execute();
    }

    public static class MyAvg{
        public Integer sum = 0;
        public Long count = 0L;
        public Double avg(){
            return sum * 1.0 / count;
        }
    }
}
